package yuekao6.tm2;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import yuekao6.util.KafkaUtil;

public class FilkSqlJoin {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //1.将order_cargo的数据与order_info数据流通过订单id进行关联。（由于存在数据乱序或者延时的情况，导致两个流的数据进度不一致
        // ，请酌情使用合适的join方式，二者数据前后相差最大5秒，同时order_info状态的ttl设置为5s）（10分）
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.kafkaSources("tms_ods"));
//        streamSource.print();

        tEnv.executeSql("CREATE TABLE tmsods (\n" +
                "  `database` STRING,\n" +
                "  `type` STRING,\n" +
                "  `table` STRING,\n" +
                "  `data` MAP<STRING,STRING>\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'tms_ods',\n" +
                "  'properties.bootstrap.servers' = 'hadoop-single:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'json'\n" +
                ")");

        Table ordercargo = tEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['order_no'] order_no,\n" +
                "`data`['status'] status,\n" +
                "`data`['collect_type'] collect_type,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['receiver_complex_id'] receiver_complex_id,\n" +
                "`data`['receiver_province_id'] receiver_province_id,\n" +
                "`data`['receiver_city_id'] receiver_city_id,\n" +
                "`data`['receiver_district_id'] receiver_district_id,\n" +
                "`data`['receiver_address'] receiver_address,\n" +
                "`data`['receiver_name'] receiver_name,\n" +
                "`data`['receiver_phone'] receiver_phone,\n" +
                "`data`['receive_location'] receive_location,\n" +
                "`data`['sender_complex_id'] sender_complex_id,\n" +
                "`data`['sender_province_id'] sender_province_id,\n" +
                "`data`['sender_city_id'] sender_city_id,\n" +
                "`data`['sender_district_id'] sender_district_id,\n" +
                "`data`['sender_address'] sender_address,\n" +
                "`data`['sender_name'] sender_name,\n" +
                "`data`['sender_phone'] sender_phone,\n" +
                "`data`['send_location'] send_location,\n" +
                "`data`['payment_type'] payment_type,\n" +
                "`data`['cargo_num'] cargo_num,\n" +
                "`data`['amount'] amount,\n" +
                "`data`['estimate_arrive_time'] estimate_arrive_time,\n" +
                "`data`['distance'] distance,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['update_time'] update_time,\n" +
                "`data`['is_deleted'] is_deleted\n" +
                " from tmsods where `table`='order_info'");


        Table orderinfo = tEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['cargo_type'] cargo_type,\n" +
                "`data`['volume_length'] volume_length,\n" +
                "`data`['volume_width'] volume_width,\n" +
                "`data`['volume_height'] volume_height,\n" +
                "`data`['weight'] weight,\n" +
                "`data`['remark'] remark,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['update_time'] update_time,\n" +
                "`data`['is_deleted'] is_deleted\n" +
                " from tmsods where `table`='order_cargo'");



        tEnv.createTemporaryView("ordercargods", ordercargo);
        tEnv.createTemporaryView("orderinfods", orderinfo);

//        tEnv.sqlQuery("select * from ordercargods").execute().print();
//        tEnv.sqlQuery("select * from orderinfods").execute().print();

        tEnv.sqlQuery("SELECT * FROM ordercargods\n" +
                "JOIN orderinfods\n" +
                "ON orderinfods.order_id = ordercargods.id ").execute().print();

/*
select
`data`['id'] id,
`data`['order_id'] order_id,
`data`['cargo_type'] cargo_type,
`data`['volume_length'] volume_length,
`data`['volume_width'] volume_width,
`data`['volume_height'] volume_height,
`data`['weight'] weight,
`data`['remark'] remark,
`data`['create_time'] create_time,
`data`['update_time'] update_time,
`data`['is_deleted'] is_deleted
 from where `table`='order_info'
 */


        env.execute();
    }
}
